package com.ferry.transport.netty.consumer;

import com.ferry.transport.api.channel.WrappedChannelGroup;
import com.ferry.transport.netty.channel.NettyChannel;
import com.ferry.transport.netty.handler.client.HandlerHolder;
import com.ferry.transport.netty.provider.DefaultProviderProcessor;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * Created by daiyong
 */
@ChannelHandler.Sharable
public abstract class ConnectionKeeper extends ChannelInboundHandlerAdapter implements TimerTask, HandlerHolder {

	private static final Logger logger = LoggerFactory.getLogger(DefaultProviderProcessor.class);

	private ChannelHandler[] handlers;

	private static final int ST_STARTED = 1;
	private static final int ST_STOPPED = 2;
	private Bootstrap bootstrap;
	private Timer timer;
	private SocketAddress remoteAddress;
	private WrappedChannelGroup group;
	private volatile int state = ST_STARTED;
	private int attempts;

	public ConnectionKeeper(Bootstrap bootstrap, Timer timer, SocketAddress remoteAddress, WrappedChannelGroup group) {
		this.bootstrap = bootstrap;
		this.timer = timer;
		this.remoteAddress = remoteAddress;
		this.group = group;
	}

	public boolean isStarted() {
		return state == ST_STARTED;
	}

	public void start() {
		state = ST_STARTED;
	}

	public void stop() {
		state = ST_STOPPED;
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		Channel ch = ctx.channel();

		if (group != null) {
			group.add(NettyChannel.attachChannel(ch));
		}

		attempts = 0;

		ctx.fireChannelActive();
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		boolean doReconnect = isReconnectNeeded();
		if (doReconnect) {
			if (attempts < 12) {
				attempts++;
			}
			long timeout = 2 << attempts;
			timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
		}
		ctx.fireChannelInactive();
	}


	@Override
	public void run(Timeout timeout) throws Exception {

		logger.info("ConnectionKeeper reconnect, remoteAddress:{}", remoteAddress);

		if (!isReconnectNeeded()) {
			return;
		}

		ChannelFuture future;
		synchronized (bootstrap) {
			bootstrap.handler(new ChannelInitializer<Channel>() {

				@Override
				protected void initChannel(Channel ch) throws Exception {
					ch.pipeline().addLast(handlers());
				}
			});
			future = bootstrap.connect(remoteAddress);
		}

		future.addListener((ChannelFutureListener) f -> {
			boolean succeed = f.isSuccess();
			if (!succeed) {
				f.channel().pipeline().fireChannelInactive();
			}
		});
	}

	private boolean isReconnectNeeded() {
		return isStarted() && (group == null || (group.size() < group.getCapacity()));
	}

}
